package com.settlement.system.handler.xxljob;

import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.settlement.system.common.constant.KafkaConstants;
import com.settlement.system.common.constant.SystemConstants;
import com.settlement.system.common.enums.LogRecordEnum;
import com.settlement.system.common.enums.PlatFormEnum;
import com.settlement.system.common.enums.StoreStatusEnum;
import com.settlement.system.common.model.Option;
import com.settlement.system.common.model.QyWechatData;
import com.settlement.system.common.util.*;
import com.settlement.system.handler.disruptor.MessageVo;
import com.settlement.system.mapper.*;
import com.settlement.system.model.entity.*;
import com.settlement.system.model.query.*;
import com.settlement.system.model.vo.BsOmsOrderImportVo;
import com.settlement.system.service.*;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import feign.Feign;
import feign.jackson.JacksonDecoder;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;

import java.lang.reflect.Field;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.concurrent.CompletableFuture;

@Component
@RequiredArgsConstructor
public class XxlJobSampleHandler {

    @Value("${xxl.job.admin.addresses}")
    private String xxlJobUrl;

    @Value("${system.config.customId}")
    private String customId;

    private final static Logger logger = LoggerFactory.getLogger(XxlJobSampleHandler.class);

    private final SysLogRecordService logRecordService;

    private final SysStoreService storeService;

    private final SysPlatformService platformService;

    private final SysStoreAccountService storeAccountService;

    private final SysDictService dictService;

    private final BsOmsOrderDetailMapper omsOrderDetailMapper;

    private final BsOmsOrderMapper omsOrderMapper;

    private final RedisTemplate redisTemplate;

    private final BsOmsReturnOrderService omsReturnOrderService;

    private final BsOmsReturnOrderDetailService omsReturnOrderDetailService;

    private final BsOmsReturnOrderSyncRecordService omsReturnOrderSyncRecordService;

    private final PlatformSourceDataService platformSourceDataService;

    private final DataSourceTransactionManager dataSourceTransactionManager;

    private final TransactionDefinition transactionDefinition;

    private final static SnowflakeIdWorker snowflakeIdWorker = SnowflakeIdWorker.getInstance();

    private final KafkaConsumeRecordService kafkaConsumeRecordService;

    private final SysBaseColumnMatchService sysBaseColumnMatchService;

    private final DyBillMapper dyBillMapper;
    private final DyFundFlowMapper dyFundFlowMapper;

    private final JdBillMapper jdBillMapper;
    private final JdFundFlowMapper jdFundFlowMapper;

    private final PddDepositMapper pddDepositMapper;
    private final PddFundFlowMapper pddFundFlowMapper;

    private final TmBillMapper tmBillMapper;
    private final TmFundFlowMapper tmFundFlowMapper;

    private final WdFundFlowMapper wdFundFlowMapper;

    private final WmFundFlowMapper wmFundFlowMapper;

    private final XhsBillMapper xhsBillMapper;
    private final XhsRefundMapper xhsRefundMapper;
    private final XhsSaleDetailMapper xhsSaleDetailMapper;
    private final XhsWechatDetailMapper xhsWechatDetailMapper;

    private final YzFundFlowMapper yzFundFlowMapper;
    private final KafkaTemplate<String, String> kafkaTemplate;

    private final KafkaSendErrorRecordService kafkaSendErrorRecordService;

    /**
     * 检查超时任务并提醒
     */
    @XxlJob("checkJobHandler")
    public void checkJobHandler() {
        logger.info("检查超时任务并提醒");
        long jobId = XxlJobHelper.getJobId();
        // 设置5秒的连接超时和10秒的读取超时
        XxlJobApi xxlJobApi = Feign.builder().decoder(new JacksonDecoder()).options(new feign.Request.Options(5000, 10000)).target(XxlJobApi.class, xxlJobUrl);
        // 通过JobId获取Job详情
        ReturnT<XxlJobInfo> info = xxlJobApi.getInfo(String.valueOf(jobId));
        String jobParam = XxlJobHelper.getJobParam().strip();
        logger.info("当前配置的超时时间为:【{}】小时", jobParam);
        LocalDateTime now = LocalDateTime.now();
        // 查询今天创建的日志记录
        List<SysLogRecord> logRecords = logRecordService.list(new LambdaQueryWrapper<SysLogRecord>()
                .ge(SysLogRecord::getCreateTime, now.toLocalDate().atStartOfDay())
                .lt(SysLogRecord::getCreateTime, now.plusDays(1).toLocalDate().atStartOfDay())
                .eq(SysLogRecord::getSend, 0));
        // 需要移除的key
        List<String> removeKeyList = new ArrayList<>();
        List<SysLogRecord> alertDataList = new ArrayList<>();
        for (SysLogRecord logRecord : logRecords) {
            logger.info("检查当前任务是否超出时间：【{}】", logRecord);
            // 超时
            if (logRecord.getCreateTime().plusHours(NumberUtils.convertInteger(jobParam)).isBefore(now)) {
                logger.info("当前任务超时，需要提醒");
                removeKeyList.add(logRecord.getJobKey());
                alertDataList.add(logRecord);
            }
            // 运行失败
            if (Objects.equals(logRecord.getResult(), LogRecordEnum.ERROR.getLabel())) {
                logger.info("当前任务出错，需要提醒");
                removeKeyList.add(logRecord.getJobKey());
                alertDataList.add(logRecord);
            }
        }
        if (removeKeyList.size() > 0) {
            logger.info("需要移除的key：【{}】", String.join(",", removeKeyList));
        }
        if (alertDataList.size() > 0) {
            List<Option> platformOptionList = dictService.listDictOptions(SystemConstants.PLATFORM_KEY);
            String content;
            QyWechatData qyWechatData;
            SysStore store;
            Option option;
            for (SysLogRecord logRecord : alertDataList) {
                store = storeService.getById(logRecord.getStoreId());
                String platformKey = store.getStoreKey();
                option = platformOptionList.stream().filter(item -> Objects.equals(item.getValue(), platformKey)).findFirst().orElse(null);
                content = "百盛RPA提醒：\n平台：<font color=\"warning\">%s</font> 店铺：<font color=\"warning\">%s</font>，RPA 运行失败。\n"
                        .formatted(option.getLabel(), store.getCommercePlatformStoreName()) +
                        "单据类型：<font color=\"comment\">%s</font>\n".formatted(logRecord.getOrderType()) +
                        "登录账号：<font color=\"comment\">%s</font>\n".formatted(logRecord.getUserName()) +
                        "<font color=\"warning\">请立即查看</font>\n";
                if (Objects.equals(logRecord.getResult(), LogRecordEnum.NOT_RUN.getLabel())) {
                    content += "<font color=\"warning\">提示：流程未运行</font>";
                } else if (Objects.equals(logRecord.getResult(), LogRecordEnum.ERROR.getLabel())) {
                    content += "<font color=\"warning\">提示：流程运行失败</font>\n<font color=\"warning\">开始执行时间：%s</font>".formatted(DateUtils.getDateString(logRecord.getStartTime(), DateUtils.crossDf));
                }
                qyWechatData = new QyWechatData(content);
                // 避免发送失败，重试3次
                for (int i = 0; i < 3; i++) {
                    try {
                        HttpUtils.sendToQyWechat(qyWechatData, info.getContent().getAlarmEmail().strip());
                        // 更新状态
                        logRecord.setSend(1);
                        logRecord.cutString();
                        logRecordService.updateById(logRecord);
                        break;
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }

    private void sendWarning(String content) {
        QyWechatData qyWechatData = new QyWechatData(content);
        String webhook = "https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=af8cd1a5-b5f0-4e72-8d6a-b20632f1d620";
        // 避免发送失败，重试3次
        for (int i = 0; i < 3; i++) {
            try {
                HttpUtils.sendToQyWechat(qyWechatData, webhook);
                break;
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    @XxlJob("createJobHandler")
    public void createJobHandler() {
        String jobParam = XxlJobHelper.getJobParam();
        if (!StringUtils.hasText(jobParam)) {
            logger.error("参数为空，不执行任务");
            return;
        }
        long jobId = XxlJobHelper.getJobId();
        // 设置5秒的连接超时和10秒的读取超时
        XxlJobApi xxlJobApi = Feign.builder().decoder(new JacksonDecoder()).options(new feign.Request.Options(5000, 10000)).target(XxlJobApi.class, xxlJobUrl);
        // 通过JobId获取Job详情
        ReturnT<XxlJobInfo> info = xxlJobApi.getInfo(String.valueOf(jobId));
        logger.info("生成【{}】定时任务", jobParam.strip());
        String storeKey;
        jobParam = jobParam.strip();
        if (jobParam.contains("：")) {
            jobParam = jobParam.replaceAll("：", ":");
        }
        String storeIdStr = null;
        if (jobParam.contains(":")) {
            storeKey = jobParam.split(":")[0];
            storeIdStr = jobParam.split(":")[1].strip();
        } else {
            storeKey = jobParam;
        }
        logger.info("报警邮件的参数【{}】", info.getContent().getAlarmEmail());
        List<SysPlatform> platformList = platformService.list();
        List<SysStore> sysStores = storeService.list(new LambdaQueryWrapper<SysStore>().eq(SysStore::getStoreKey, storeKey.strip()));
        if (StringUtils.hasText(storeIdStr)) {
            logger.info("正在过滤店铺：【{}】", storeIdStr);
            storeIdStr = storeIdStr.replaceAll("，", ",");
            List<Long> storeIds = new ArrayList<>();
            for (String idStr : storeIdStr.split(",")) {
                storeIds.add(Long.valueOf(idStr));
            }
            sysStores = sysStores.stream().filter(item -> storeIds.contains(item.getId())).toList();
            logger.info("过滤后的店铺：【{}】", sysStores);
        }
        if (sysStores.size() == 0) {
            logger.error("店铺为空，不执行任务");
            return;
        }
        List<String> codeList = sysStores.stream().filter(item -> Objects.equals(item.getStatus(), StoreStatusEnum.ENABLE.getValue())).map(SysStore::getCommercePlatformStoreCode).filter(StringUtils::hasText).toList();
        logger.info("启用的店铺code列表：【{}】", String.join(",", codeList));
        List<SysStoreAccount> sysStoreAccountList = storeAccountService.list(new LambdaQueryWrapper<SysStoreAccount>().in(SysStoreAccount::getCommercePlatformStoreCode, codeList));
        SysStore store;
        SysPlatform platform;
        String now = LocalDate.now().toString();
        logger.info("当前日期：【{}】", now);
        SysLogRecord logRecord;
        if (StringUtils.hasText(info.getContent().getAlarmEmail())) {
            logger.info("过滤任务类型：【{}】", info.getContent().getAlarmEmail());
            sysStoreAccountList.removeIf(item -> !Objects.equals(item.getAccountType(), info.getContent().getAlarmEmail().strip()));
        }
        String userName;
        for (SysStoreAccount storeAccount : sysStoreAccountList) {
            logger.info("当前店铺账号信息：【{}】", storeAccount);
            store = sysStores.stream().filter(item -> Objects.equals(item.getCommercePlatformStoreCode(), storeAccount.getCommercePlatformStoreCode())).findFirst().orElse(null);
            if (Objects.nonNull(store)) {
                logger.info("当前店铺：【{}】", store);
                userName = SimpleEncryptHelper.decode(storeAccount.getUserName());
                // 拼接方式为 平台KEY:店铺id:当前日期:店铺账号:店铺标识:账号类型
                String redisKey = "%s:%s:%s:%s:%s:%s".formatted(store.getStoreKey(), store.getId(), now, userName, storeAccount.getAccount(), storeAccount.getAccountType());
                logger.info("生成的key：【{}】", redisKey);
                logRecord = new SysLogRecord();
                logRecord.setStoreId(store.getId());
                String platformKey = store.getStoreKey();
                platform = platformList.stream().filter(item -> Objects.equals(item.getPlatformKey(), platformKey)).findFirst().orElse(null);
                if (Objects.nonNull(platform)) {
                    logRecord.setPlatformId(platform.getId());
                }
                logRecord.setOrderType(storeAccount.getAccountType());
                logRecord.setUserName(userName);
                // 任务KEY
                logRecord.setJobKey(redisKey);
                logRecordService.save(logRecord);
                logger.info("保存的任务日志信息：【{}】", logRecord);
            }
        }
    }

    @XxlJob("jstOrderHandler")
    public void jstOrderHandler() throws Exception {
        // 删除缓存
        redisTemplate.delete(SystemConstants.OMS_ORDER_KEY);
        logger.info("开始执行 JST 订单同步任务...");
        String jobParam = XxlJobHelper.getJobParam().strip();
        // 格式为： 2024-06-25 00:00:00,2024-07-01 00:00:00
        LocalDateTime start;
        LocalDateTime end;
        if (StringUtils.hasText(jobParam)) {
            String[] dates = jobParam.split(",");
            start = LocalDateTime.parse(dates[0].strip(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            end = LocalDateTime.parse(dates[1].strip(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        } else {
            start = LocalDateTime.now().minusDays(1);
            end = LocalDateTime.now();
        }
        List<DateItem> dateItemList = new ArrayList<>();
        // 组装日期列表
        while (start.plusDays(1).isBefore(end)) {
            DateItem dateItem = new DateItem();
            dateItem.setStartTime(start);
            dateItem.setEndTime(start.plusDays(1));
            dateItemList.add(dateItem);
            start = start.plusDays(1);
        }
        // 加上最后一个时间段
        dateItemList.add(new DateItem(start, end));
        String startStr;
        String endStr;
        int pageSize = 100;
        String dataType = "订单";
        BsOmsReturnOrderSyncRecord syncRecord;
        List<BsOmsReturnOrderSyncRecord> syncRecords;
        logger.info("开始时间：{}", start);
        logger.info("结束时间：{}", end);
        List<SysStore> storeList = storeService.list();
        Map<String, Object> params;
        // 通过配置获取账单月份【billMonth】
        List<SysBaseColumnMatch> configListByPlatform = sysBaseColumnMatchService.getAllConfigListByPlatform(SystemConstants.OMS_ORDER_DICT_KEY);
        SysBaseColumnMatch baseColumnMatch;
        for (DateItem dateItem : dateItemList) {
            try {
                syncRecords = omsReturnOrderSyncRecordService.list(new LambdaQueryWrapper<BsOmsReturnOrderSyncRecord>().eq(BsOmsReturnOrderSyncRecord::getDataType, dataType).eq(BsOmsReturnOrderSyncRecord::getSyncDate, dateItem.getStartTime().toLocalDate()));
                if (syncRecords.size() > 0) {
                    logger.info("此日期【{}】已经同步完成，跳过...", dateItem.getStartTime().toLocalDate().toString());
                    continue;
                }
                logger.info("查询此日期【{}】是否已经同步完成，结果：{}", dateItem.getStartTime().toLocalDate().toString(), syncRecords.size() > 0);
                int pageIndex = 1;
                int pageCount = 0;
                logger.info("正在查询开始时间：{}", dateItem.getStartTime().toString());
                logger.info("正在查询结束时间：{}", dateItem.getEndTime().toString());
                params = new HashMap<>();
                startStr = dateItem.getStartTime().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                endStr = dateItem.getEndTime().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                params.put("start_time", startStr);
                params.put("end_time", endStr);
                params.put("date_type", "3");
                params.put("page_index", pageIndex);
                params.put("status", "Sent");
                params.put("page_size", pageSize);
                logger.info("查询参数：【{}】", params);
                JSONObject result = HttpUtils.invokeJSTApiForJSON(customId, "jushuitan.order.list.query", params);
                if (Objects.nonNull(result)) {
                    JSONObject order;
                    List<BsOmsOrder> omsOrders;
                    List<BsOmsOrderDetail> omsOrderDetails;
                    BsOmsOrder omsOrder;
                    BsOmsOrderDetail omsOrderDetail;
                    String tableName;
                    String created;
                    SysStore store;
                    JSONArray items;
                    JSONArray orders = result.getJSONObject("response").getJSONArray("orders");
                    logger.info("查询结果：共有【{}】条数据...", result.getJSONObject("response").getInteger("data_count"));
                    pageCount = result.getJSONObject("response").getInteger("page_count");
                    logger.info("查询结果：共有【{}】页数据...", pageCount);
                    while (Objects.nonNull(orders) && orders.size() > 0) {
                        logger.info("正在处理第{}页...", pageIndex);
                        // 循环处理
                        omsOrders = new ArrayList<>();
                        for (int i = 0; i < orders.size(); i++) {
                            order = orders.getJSONObject(i);
                            long shopId = order.getLong("shop_id");
                            store = storeList.stream().filter(item -> Objects.equals(item.getInnerSystemStoreCode(), shopId + "")).findFirst().orElse(null);
                            if (Objects.isNull(store)) {
                                logger.info("店铺不存在，shopId：{}", shopId);
                                continue;
                            }
                            created = order.getString("created");
                            omsOrder = new BsOmsOrder();
                            omsOrderDetails = new ArrayList<>();
                            omsOrder.setCreated(DateUtils.convertStringToLocalDateTime(created));
                            // 店铺
                            omsOrder.setStoreCode(store.getInnerSystemStoreCode());
                            // 内部订单号
                            omsOrder.setOId(order.getInteger("o_id").toString());
                            // 买家实付金额
                            if (order.containsKey("buyer_paid_amount")) {
                                omsOrder.setBuyerPaidAmount(NumberUtils.convertBigDecimal(order.getString("buyer_paid_amount")));
                            }
                            // 买家实付金额
                            if (order.containsKey("seller_income_amount")) {
                                omsOrder.setSellerIncomeAmount(NumberUtils.convertBigDecimal(order.getString("seller_income_amount")));
                            }
                            // 线上订单编号
                            if (order.containsKey("merge_so_id")) {
                                omsOrder.setMergeSoId(order.getString("merge_so_id"));
                            }
                            // 下单时间
                            omsOrder.setOrderDate(DateUtils.convertStringToLocalDateTime(order.getString("order_date")));
                            // 付款日期
                            omsOrder.setPayDate(DateUtils.convertStringToLocalDateTime(order.getString("pay_date")));
                            // 发货日期
                            omsOrder.setSendDate(DateUtils.convertStringToLocalDateTime(order.getString("send_date")));
                            // 应付金额
                            omsOrder.setPayAmount(NumberUtils.convertBigDecimal(order.getString("pay_amount")));
                            // 已付金额
                            omsOrder.setPaidAmount(NumberUtils.convertBigDecimal(order.getString("paid_amount")));
                            // 状态
                            omsOrder.setStatus(order.getString("status"));
                            try {
                                // 支付单号
                                omsOrder.setOuterPayId(order.getJSONArray("pays").getJSONObject(0).getString("outer_pay_id"));
                                // 商品总成交金额
                                omsOrder.setAmount(NumberUtils.convertBigDecimal(order.getJSONArray("pays").getJSONObject(0).getString("amount")));
                            } catch (Exception e) {
                                e.printStackTrace();
                            }
                            if (order.containsKey("items")) {
                                items = order.getJSONArray("items");
                                if (Objects.nonNull(items) && items.size() > 0) {
                                    // 循环处理明细
                                    for (int j = 0; j < items.size(); j++) {
                                        JSONObject item = items.getJSONObject(j);
                                        omsOrderDetail = new BsOmsOrderDetail();
                                        // 买家实付金额
                                        try {
                                            omsOrderDetail.setBuyerPaidAmount(NumberUtils.convertBigDecimal(item.getString("buyer_paid_amount")));
                                        } catch (Exception e) {
                                            e.printStackTrace();
                                        }
                                        // 卖家实收金额
                                        try {
                                            omsOrderDetail.setSellerIncomeAmount(NumberUtils.convertBigDecimal(item.getString("seller_income_amount")));
                                        } catch (Exception e) {
                                            e.printStackTrace();
                                        }
                                        // 原始线上订单号
                                        omsOrderDetail.setRawSoId(item.getString("raw_so_id"));
                                        // 线上子订单编号
                                        omsOrderDetail.setOuterOiId(item.getString("outer_oi_id"));
                                        // 子订单编号
                                        omsOrderDetail.setOiId(item.getString("oi_id"));
                                        // 商品编码
                                        omsOrderDetail.setSkuId(item.getString("sku_id"));
                                        // 数量
                                        omsOrderDetail.setQty(item.getInteger("qty"));
                                        // 商品单价
                                        omsOrderDetail.setPrice(NumberUtils.convertBigDecimal(item.getString("price")));
                                        // 原价
                                        omsOrderDetail.setBasePrice(NumberUtils.convertBigDecimal(item.getString("base_price")));
                                        omsOrderDetails.add(omsOrderDetail);
                                    }
                                }
                            }
                            omsOrder.setDetailList(omsOrderDetails);
                            omsOrders.add(omsOrder);
                        }
                        if (omsOrders.size() > 0) {
                            Map<String, String> dataMap;
                            Map<String, String> itemDataMap;
                            MessageVo vo;
                            Field declaredField;
                            String linkNoSourceValue = null;
                            String mainNoSourceValue = null;
                            String subNoSourceValue = null;
                            for (BsOmsOrder originalOmsOrder : omsOrders) {
                                tableName = "bs_oms_order_%s".formatted(DateUtils.formatDate(originalOmsOrder.getSendDate()));
                                originalOmsOrder.setTableName(tableName);
                                originalOmsOrder.setId(snowflakeIdWorker.nextId(tableName.replaceAll("bs_oms_order_", "")));
                                if (!StringUtils.hasText(omsOrderMapper.existTable(tableName))) {
                                    try {
                                        omsOrderMapper.createTable(tableName);
                                        // 同时创建明细表
                                        omsOrderMapper.createDetailTable("bs_oms_order_detail_%s".formatted(tableName.replaceAll("bs_oms_order_", "")));
                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                }
                                omsOrderMapper.insertData(originalOmsOrder);
                                // 保存一份OMS销售订单对应表记录
                                baseColumnMatch = configListByPlatform.stream().filter(item -> Objects.equals(SystemConstants.LINK_NO, item.getStandardColumn())).findFirst().orElse(null);
                                if (Objects.nonNull(baseColumnMatch)) {
                                    if (BsOmsOrderImportVo.getKeyMap().containsKey(baseColumnMatch.getSourceColumn())) {
                                        String sourceEn = BsOmsOrderImportVo.getKeyMap().get(baseColumnMatch.getSourceColumn());
                                        // 通过反射获取原始值
                                        try {
                                            declaredField = originalOmsOrder.getClass().getDeclaredField(sourceEn);
                                            declaredField.setAccessible(true);
                                            linkNoSourceValue = (String) declaredField.get(originalOmsOrder);
                                        } catch (Exception ex) {
                                            linkNoSourceValue = null;
                                        }
                                    }
                                }
                                baseColumnMatch = configListByPlatform.stream().filter(item -> Objects.equals(SystemConstants.MAIN_NO, item.getStandardColumn())).findFirst().orElse(null);
                                if (Objects.nonNull(baseColumnMatch)) {
                                    if (BsOmsOrderImportVo.getKeyMap().containsKey(baseColumnMatch.getSourceColumn())) {
                                        String sourceEn = BsOmsOrderImportVo.getKeyMap().get(baseColumnMatch.getSourceColumn());
                                        // 通过反射获取原始值
                                        try {
                                            declaredField = originalOmsOrder.getClass().getDeclaredField(sourceEn);
                                            declaredField.setAccessible(true);
                                            mainNoSourceValue = (String) declaredField.get(originalOmsOrder);
                                        } catch (Exception ex) {
                                            mainNoSourceValue = null;
                                        }
                                    }
                                }
                                baseColumnMatch = configListByPlatform.stream().filter(item -> Objects.equals(SystemConstants.SUB_NO, item.getStandardColumn())).findFirst().orElse(null);
                                if (Objects.nonNull(baseColumnMatch)) {
                                    if (BsOmsOrderImportVo.getKeyMap().containsKey(baseColumnMatch.getSourceColumn())) {
                                        String sourceEn = BsOmsOrderImportVo.getKeyMap().get(baseColumnMatch.getSourceColumn());
                                        // 通过反射获取原始值
                                        try {
                                            declaredField = originalOmsOrder.getClass().getDeclaredField(sourceEn);
                                            declaredField.setAccessible(true);
                                            subNoSourceValue = (String) declaredField.get(originalOmsOrder);
                                        } catch (Exception ex) {
                                            subNoSourceValue = null;
                                        }
                                    }
                                }
                                if (Objects.nonNull(linkNoSourceValue) && Objects.nonNull(mainNoSourceValue) && Objects.nonNull(subNoSourceValue)) {
                                    platformSourceDataService.saveOmsLink(linkNoSourceValue, mainNoSourceValue, subNoSourceValue, originalOmsOrder.getId());
                                }
                                // logger.info("生成百盛订单：{}", originalOmsOrder);
                                if (Objects.nonNull(originalOmsOrder.getDetailList()) && originalOmsOrder.getDetailList().size() > 0) {
                                    for (BsOmsOrderDetail bsOmsOrderDetail : originalOmsOrder.getDetailList()) {
                                        bsOmsOrderDetail.setOId(originalOmsOrder.getId());
                                        bsOmsOrderDetail.setTableName("bs_oms_order_detail_%s".formatted(originalOmsOrder.getTableName().replaceAll("bs_oms_order_", "")));
                                        omsOrderDetailMapper.insertData(bsOmsOrderDetail);
                                        // logger.info("生成明细数据：{}", bsOmsOrderDetail);
                                        // 发送消息
                                        vo = new MessageVo();
                                        dataMap = new HashMap<>();
                                        objectToMap(originalOmsOrder, dataMap);
                                        itemDataMap = new HashMap<>();
                                        objectToMap(bsOmsOrderDetail, itemDataMap);
                                        itemDataMap.remove("oId");
                                        for (String itemKey : itemDataMap.keySet()) {
                                            // 明细数据 覆盖 主表数据
                                            dataMap.put(itemKey, itemDataMap.get(itemKey));
                                        }
                                        // PID
                                        dataMap.put("pId", bsOmsOrderDetail.getOiId());
                                        dataMap.remove("detailList");
                                        // 是否是OMS订单
                                        vo.setOmsOrder(true);
                                        store = storeService.getByInnerSystemStoreCode(originalOmsOrder.getStoreCode());
                                        vo.setStoreId(store.getId());
                                        // 是否是OMS售后单
                                        vo.setOmsReturnOrder(false);
                                        vo.setDataMap(dataMap);
                                        vo.setUuid(snowflakeIdWorker.nextId());
                                        // 推送到转换标准订单的消息队列
                                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_ORDER_QUEUE, JSONObject.toJSONString(vo));
                                        MessageVo finalVo = vo;
                                        completableFuture.whenComplete((sendResult, throwable) -> {
                                            if (Objects.nonNull(throwable)) {
                                                throwable.printStackTrace();
                                                logger.info("接口推送OMS订单消息发送失败：{}", throwable.getMessage());
                                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_ORDER_QUEUE);
                                                errorRecord.setRemark(throwable.getMessage());
                                                errorRecord.setUuid(finalVo.getUuid());
                                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                                            }
                                        });
                                    }
                                }
                            }
                        }
                        try {
                            Thread.sleep(10000);
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        pageIndex += 1;
                        logger.info("查询{}页，共{}页...", pageIndex, pageCount);
                        params = new HashMap<>();
                        params.put("start_time", startStr);
                        params.put("end_time", endStr);
                        params.put("date_type", "3");
                        params.put("page_index", pageIndex);
                        params.put("page_size", pageSize);
                        params.put("status", "Sent");
                        logger.info("查询参数：【{}】", params);
                        for (int i = 0; i < 3; i++) {
                            try {
                                result = HttpUtils.invokeJSTApiForJSON(customId, "jushuitan.order.list.query", params);
                                break;
                            } catch (Exception ex) {
                                ex.printStackTrace();
                                try {
                                    Thread.sleep(3000);
                                } catch (Exception exception) {
                                }
                            }
                        }
                        orders = result.getJSONObject("response").getJSONArray("orders");
                    }
                }
                logger.info("正在保存一份同步记录：{}", dateItem.getStartTime().toLocalDate().toString());
                syncRecord = new BsOmsReturnOrderSyncRecord();
                syncRecord.setDataType(dataType);
                syncRecord.setSyncDate(dateItem.getStartTime().toLocalDate());
                omsReturnOrderSyncRecordService.save(syncRecord);
            } catch (Exception e) {
                e.printStackTrace();
                logger.info("JST 订单同步任务执行出错...");
            }
        }
        logger.info("JST 订单同步任务执行完成...");
    }

    public static void objectToMap(Object object, Map<String, String> dataMap) {
        Class<?> clazz = object.getClass();
        for (Field field : clazz.getDeclaredFields()) {
            try {
                field.setAccessible(true);
                if (Objects.nonNull(field.get(object))) {
                    dataMap.put(field.getName(), field.get(object).toString());
                }
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }
        }
    }

    @XxlJob("jstReturnOrderHandler")
    public void jstReturnOrderHandler() throws Exception {
        logger.info("JST 退货订单同步任务开始执行...");
        String jobParam = XxlJobHelper.getJobParam().strip();
        // 格式为： 2024-06-25 00:00:00,2024-07-01 00:00:00
        LocalDateTime start;
        LocalDateTime end;
        if (StringUtils.hasText(jobParam)) {
            String[] dates = jobParam.split(",");
            start = LocalDateTime.parse(dates[0].strip(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            end = LocalDateTime.parse(dates[1].strip(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        } else {
            start = LocalDateTime.now().minusDays(1);
            end = LocalDateTime.now();
        }
        String startStr;
        String endStr;
        int pageSize = 100;
        logger.info("开始时间：{}", start);
        logger.info("结束时间：{}", end);
        List<DateItem> dateItemList = new ArrayList<>();
        // 组装日期列表
        while (start.plusDays(1).isBefore(end)) {
            DateItem dateItem = new DateItem();
            dateItem.setStartTime(start);
            dateItem.setEndTime(start.plusDays(1));
            dateItemList.add(dateItem);
            start = start.plusDays(1);
        }
        // 加上最后一个时间段
        dateItemList.add(new DateItem(start, end));
        List<SysStore> storeList = storeService.list();
        Map<String, Object> params;
        BsOmsReturnOrderSyncRecord syncRecord;
        List<BsOmsReturnOrderSyncRecord> syncRecords;
        String dataType = "售后单";
        for (DateItem dateItem : dateItemList) {
            // 这个用另一种事务方式，查询一天就提交一天的事务
            TransactionStatus transactionStatus = null;
            try {
                // 开启事务
                transactionStatus = dataSourceTransactionManager.getTransaction(transactionDefinition);
                logger.info("正在查询开始时间：{}", dateItem.getStartTime().toString());
                logger.info("正在查询结束时间：{}", dateItem.getEndTime().toString());
                startStr = dateItem.getStartTime().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                endStr = dateItem.getEndTime().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
                syncRecords = omsReturnOrderSyncRecordService.list(new LambdaQueryWrapper<BsOmsReturnOrderSyncRecord>().eq(BsOmsReturnOrderSyncRecord::getDataType, dataType).eq(BsOmsReturnOrderSyncRecord::getSyncDate, dateItem.getStartTime().toLocalDate()));
                if (syncRecords.size() > 0) {
                    logger.info("此日期【{}】已经同步完成，跳过...", dateItem.getStartTime().toLocalDate().toString());
                    continue;
                }
                logger.info("查询此日期【{}】是否已经同步完成，结果：{}", dateItem.getStartTime().toLocalDate().toString(), syncRecords.size() > 0);
                int pageIndex = 1;
                int pageCount = 0;
                params = new HashMap<>();
                params.put("start_time", startStr);
                params.put("end_time", endStr);
                params.put("date_type", "2");
                params.put("status", "Confirmed");
                params.put("page_index", pageIndex);
                params.put("page_size", pageSize);
                logger.info("查询参数：【{}】", params);
                JSONObject result = HttpUtils.invokeJSTApiForJSON(customId, "jushuitan.refund.list.query", params);
                JSONArray orders = result.getJSONObject("response").getJSONArray("datas");
                logger.info("查询结果：共有【{}】条数据...", result.getJSONObject("response").getInteger("data_count"));
                pageCount = result.getJSONObject("response").getInteger("page_count");
                logger.info("查询结果：共有【{}】页数据...", pageCount);
                BsOmsReturnOrder omsReturnOrder;
                BsOmsReturnOrderDetail omsReturnOrderDetail;
                List<BsOmsReturnOrderDetail> detailList;
                List<BsOmsReturnOrder> omsReturnOrders;
                JSONObject order;
                SysStore store;
                while (Objects.nonNull(orders) && orders.size() > 0) {
                    omsReturnOrders = new ArrayList<>();
                    detailList = new ArrayList<>();
                    logger.info("正在处理第{}页...", pageIndex);
                    for (int i = 0; i < orders.size(); i++) {
                        order = orders.getJSONObject(i);
                        long shopId = order.getLong("shop_id");
                        store = storeList.stream().filter(item -> Objects.equals(item.getInnerSystemStoreCode(), shopId + "")).findFirst().orElse(null);
                        if (Objects.isNull(store)) {
                            logger.info("店铺不存在，shopId：{}", shopId);
                            continue;
                        }
                        omsReturnOrder = new BsOmsReturnOrder();
                        omsReturnOrder.setId(snowflakeIdWorker.nextId());
                        // 店铺
                        omsReturnOrder.setStoreCode(store.getInnerSystemStoreCode());
                        // 售后单号
                        omsReturnOrder.setAsId(order.getString("as_id"));
                        // 售后类型
                        omsReturnOrder.setOrderType(order.getString("type"));
                        // 内部订单号
                        omsReturnOrder.setOId(order.getInteger("o_id").toString());
                        // 线上订单号
                        omsReturnOrder.setSoId(order.getString("so_id"));
                        // 状态
                        omsReturnOrder.setStatus(order.getString("status"));
                        // 货物状态
                        omsReturnOrder.setGoodStatus(order.getString("good_status"));
                        // 最后确认日期
                        omsReturnOrder.setConfirmDate(DateUtils.convertStringToLocalDateTime(order.getString("confirm_date")));
                        // 登记时间
                        omsReturnOrder.setCreated(DateUtils.convertStringToLocalDateTime(order.getString("created")));
                        // 申请日期
                        omsReturnOrder.setAsDate(DateUtils.convertStringToLocalDateTime(order.getString("as_date")));
                        // 线上申请金额
                        omsReturnOrder.setBuyerApplyRefund(NumberUtils.convertBigDecimal(order.getString("payment")));
                        // 线上退运费
                        omsReturnOrder.setShopFreight(NumberUtils.convertBigDecimal(order.getString("shop_freight")));
                        if (order.containsKey("free_amount")) {
                            // 优惠/差额
                            omsReturnOrder.setFreeAmount(NumberUtils.convertBigDecimal(order.getString("free_amount")));
                        }
                        if (order.containsKey("refund")) {
                            // 卖家应退金额
                            omsReturnOrder.setRefund(NumberUtils.convertBigDecimal(order.getString("refund")));
                        }
                        if (order.containsKey("payment")) {
                            // 买家应补偿金额
                            omsReturnOrder.setPayment(NumberUtils.convertBigDecimal(order.getString("payment")));
                        }
                        if (order.containsKey("urefund")) {
                            // 卖家应补
                            omsReturnOrder.setUrefund(NumberUtils.convertBigDecimal(order.getString("urefund")));
                        }
                        omsReturnOrders.add(omsReturnOrder);
                        if (order.getJSONArray("items").size() > 0) {
                            // 循环明细
                            for (int j = 0; j < order.getJSONArray("items").size(); j++) {
                                JSONObject item = order.getJSONArray("items").getJSONObject(j);
                                omsReturnOrderDetail = new BsOmsReturnOrderDetail();
                                // 订单ID
                                omsReturnOrderDetail.setOId(omsReturnOrder.getId());
                                // 售后分类
                                omsReturnOrderDetail.setOrderType(item.getString("type"));
                                // 商品编码
                                omsReturnOrderDetail.setSkuId(item.getString("sku_id"));
                                // 线上子订单编号
                                omsReturnOrderDetail.setOuterOiId(item.getString("outer_oi_id"));
                                // 申请数量
                                omsReturnOrderDetail.setQty(item.getInteger("qty"));
                                // 退货数量
                                omsReturnOrderDetail.setRQty(item.getInteger("r_qty"));
                                // 线上明细金额
                                omsReturnOrderDetail.setShopAmount(NumberUtils.convertBigDecimal(item.getString("shop_amount")));
                                // 单价
                                omsReturnOrderDetail.setPrice(NumberUtils.convertBigDecimal(item.getString("price")));
                                // 申请金额
                                omsReturnOrderDetail.setAmount(NumberUtils.convertBigDecimal(item.getString("amount")));
                                // 是否赠品
                                omsReturnOrderDetail.setIsGift(item.getString("is_gift"));
                                // 商品品牌
                                omsReturnOrderDetail.setBrand(item.getString("brand"));
                                detailList.add(omsReturnOrderDetail);
                            }
                        }
                    }
                    if (omsReturnOrders.size() > 0) {
                        omsReturnOrderService.saveBatch(omsReturnOrders);
                    }
                    if (detailList.size() > 0) {
                        omsReturnOrderDetailService.saveBatch(detailList);
                    }
                    Map<String, String> dataMap;
                    Map<String, String> itemDataMap;
                    MessageVo vo;
                    for (BsOmsReturnOrderDetail detail : detailList) {
                        omsReturnOrder = omsReturnOrders.stream().filter(o -> Objects.equals(o.getId(), detail.getOId())).findFirst().get();
                        vo = new MessageVo();
                        dataMap = new HashMap<>();
                        objectToMap(omsReturnOrder, dataMap);
                        itemDataMap = new HashMap<>();
                        objectToMap(detail, itemDataMap);
                        itemDataMap.remove("oId");
                        for (String itemKey : itemDataMap.keySet()) {
                            // 明细数据 覆盖 主表数据
                            dataMap.put(itemKey, itemDataMap.get(itemKey));
                        }
                        // PID
                        dataMap.put("pId", "%s-%s".formatted(omsReturnOrder.getId(), detail.getId()));
                        // 是否是OMS订单
                        vo.setOmsOrder(true);
                        store = storeService.getByInnerSystemStoreCode(omsReturnOrder.getStoreCode());
                        vo.setStoreId(store.getId());
                        // 是否是OMS售后单
                        vo.setOmsReturnOrder(true);
                        vo.setDataMap(dataMap);
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_ORDER_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((sendResult, throwable) -> {
                            if (Objects.nonNull(throwable)) {
                                throwable.printStackTrace();
                                logger.info("接口推送OMS售后单消息发送失败：{}", throwable.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_ORDER_QUEUE);
                                errorRecord.setRemark(throwable.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                    try {
                        Thread.sleep(10000);
                    } catch (Exception e) {
                    }
                    pageIndex += 1;
                    logger.info("查询{}页，共{}页...", pageIndex, pageCount);
                    params = new HashMap<>();
                    params.put("start_time", startStr);
                    params.put("end_time", endStr);
                    params.put("date_type", "2");
                    params.put("status", "Confirmed");
                    params.put("page_index", pageIndex);
                    params.put("page_size", pageSize);
                    logger.info("查询参数：【{}】", params);
                    for (int i = 0; i < 3; i++) {
                        try {
                            result = HttpUtils.invokeJSTApiForJSON(customId, "jushuitan.refund.list.query", params);
                            break;
                        } catch (Exception e) {
                            e.printStackTrace();
                            try {
                                Thread.sleep(3000);
                            } catch (Exception ex) {
                            }
                        }
                    }
                    orders = result.getJSONObject("response").getJSONArray("datas");
                }
                logger.info("正在保存一份同步记录：{}", dateItem.getStartTime().toLocalDate().toString());
                syncRecord = new BsOmsReturnOrderSyncRecord();
                syncRecord.setDataType(dataType);
                syncRecord.setSyncDate(dateItem.getStartTime().toLocalDate());
                omsReturnOrderSyncRecordService.save(syncRecord);
                // 手动提交事务
                dataSourceTransactionManager.commit(transactionStatus);
            } catch (Exception e) {
                e.printStackTrace();
                logger.error("JST 退货订单同步任务执行失败：{}", e.getMessage());
                // 手动回滚事务
                if (Objects.nonNull(transactionStatus)) {
                    dataSourceTransactionManager.rollback(transactionStatus);
                }
            }
        }
        logger.info("JST 退货订单同步任务执行完成...");
    }

    @XxlJob("fixBillHandler")
    public void fixBillHandler() {
        List<String> billTypeList = Arrays.asList("dyBill", "dyFundFlow", "jdBill", "jdFundFlow", "pddDeposit", "pddFundFlow", "tmBill", "tmFundFlow", "wdFundFlow", "wmFundFlow", "redBookbill", "redBookRefund", "redBookSaleDetail", "redBookWechatDetail", "yzFundFlow");
        String jobParam = XxlJobHelper.getJobParam();
        List<String> paramList;
        if (StringUtils.hasText(jobParam)) {
            paramList = List.of(jobParam.split(","));
        } else {
            paramList = billTypeList;
        }
        logger.info("开始修复【{}】账单...", String.join(",", paramList));
        for (String billType : paramList) {
            if (Objects.equals("dyBill", billType)) {
                try {
                    fixDyBill();
                } catch (Exception e) {
                    sendWarning("修复【抖音账单】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("dyFundFlow", billType)) {
                try {
                    fixDyFundFlow();
                } catch (Exception e) {
                    sendWarning("修复【抖音资金流水】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("jdBill", billType)) {
                try {
                    fixJdBill();
                } catch (Exception e) {
                    sendWarning("修复【京东账单】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("jdFundFlow", billType)) {
                try {
                    fixJdFundFlow();
                } catch (Exception e) {
                    sendWarning("修复【京东资金流水】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("pddDeposit", billType)) {
                try {
                    fixPddDeposit();
                } catch (Exception e) {
                    sendWarning("修复【拼多多保证金明细】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("pddFundFlow", billType)) {
                try {
                    fixPddFundFlow();
                } catch (Exception e) {
                    sendWarning("修复【拼多多资金流水】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("tmBill", billType)) {
                try {
                    fixTmBill();
                } catch (Exception e) {
                    sendWarning("修复【天猫账单】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("tmFundFlow", billType)) {
                try {
                    fixTmFundFlow();
                } catch (Exception e) {
                    sendWarning("修复【天猫资金流水】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("wdFundFlow", billType)) {
                try {
                    fixWdFundFlow();
                } catch (Exception e) {
                    sendWarning("修复【微店资金流水】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("wmFundFlow", billType)) {
                try {
                    fixWmFundFlow();
                } catch (Exception e) {
                    sendWarning("修复【微盟资金流水】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("redBookbill", billType)) {
                try {
                    fixXhsBill();
                } catch (Exception e) {
                    sendWarning("修复【小红书账单】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("redBookRefund", billType)) {
                try {
                    fixXhsRefund();
                } catch (Exception e) {
                    sendWarning("修复【小红书退款】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("redBookSaleDetail", billType)) {
                try {
                    fixXhsSaleDetail();
                } catch (Exception e) {
                    sendWarning("修复【小红书销售明细】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("redBookWechatDetail", billType)) {
                try {
                    fixXhsWechatDetail();
                } catch (Exception e) {
                    sendWarning("修复【小红书微信明细】出现问题：%s".formatted(e.getMessage()));
                }
            } else if (Objects.equals("yzFundFlow", billType)) {
                try {
                    fixYzFundFlow();
                } catch (Exception e) {
                    sendWarning("修复【有赞资金流水】出现问题：%s".formatted(e.getMessage()));
                }
            }
        }
    }

    private void fixYzFundFlow() {
        logger.info("开始修复有赞资金流水...");
        List<String> allTable = yzFundFlowMapper.getAllTable();
        Page<YzFundFlow> page;
        YzFundFlowQuery query;
        String dataId;
        MessageVo vo;
        for (String tableName : allTable) {
            logger.info("开始修复表：{}", tableName);
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new YzFundFlowQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            yzFundFlowMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (YzFundFlow yzFundFlow : page.getRecords()) {
                    dataId = yzFundFlow.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(yzFundFlow.getId());
                        vo.setStoreId(yzFundFlow.getStoreId());
                        vo.setBillMonth(yzFundFlow.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("yz_fund_flow_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode("yzFundFlow");
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送有赞资金流水数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                yzFundFlowMapper.getPage(page, query);
            }
        }
        logger.info("修复有赞资金流水完成");
    }

    private void fixXhsWechatDetail() {
        logger.info("开始修复小红书微信明细...");
        List<String> allTable = xhsWechatDetailMapper.getAllTable();
        Page<XhsWechatDetail> page;
        XhsWechatDetailQuery query;
        String dataId;
        MessageVo vo;
        for (String tableName : allTable) {
            logger.info("开始修复表：{}", tableName);
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new XhsWechatDetailQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            xhsWechatDetailMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (XhsWechatDetail xhsWechatDetail : page.getRecords()) {
                    dataId = xhsWechatDetail.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(xhsWechatDetail.getId());
                        vo.setStoreId(xhsWechatDetail.getStoreId());
                        vo.setBillMonth(xhsWechatDetail.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("xhs_wechat_detail_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode("redBookWechatDetail");
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送小红书微信明细数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                xhsWechatDetailMapper.getPage(page, query);
            }
        }
        logger.info("小红书微信明细修复完成");
    }

    private void fixXhsSaleDetail() {
        logger.info("开始修复小红书销售明细...");
        List<String> allTable = xhsSaleDetailMapper.getAllTable();
        Page<XhsSaleDetail> page;
        XhsSaleDetailQuery query;
        String dataId;
        MessageVo vo;
        for (String tableName : allTable) {
            logger.info("开始修复表：{}", tableName);
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new XhsSaleDetailQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            xhsSaleDetailMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (XhsSaleDetail xhsSaleDetail : page.getRecords()) {
                    dataId = xhsSaleDetail.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(xhsSaleDetail.getId());
                        vo.setStoreId(xhsSaleDetail.getStoreId());
                        vo.setBillMonth(xhsSaleDetail.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("xhs_sale_detail_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode("redBookSaleDetail");
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送小红书销售明细数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                xhsSaleDetailMapper.getPage(page, query);
            }
        }
        logger.info("小红书销售明细修复完成");
    }

    private void fixXhsRefund() {
        logger.info("开始修复小红书退款单...");
        List<String> allTable = xhsRefundMapper.getAllTable();
        Page<XhsRefund> page;
        XhsRefundQuery query;
        String dataId;
        MessageVo vo;
        for (String tableName : allTable) {
            logger.info("开始修复表：{}", tableName);
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new XhsRefundQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            xhsRefundMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (XhsRefund xhsRefund : page.getRecords()) {
                    dataId = xhsRefund.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(xhsRefund.getId());
                        vo.setStoreId(xhsRefund.getStoreId());
                        vo.setBillMonth(xhsRefund.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("xhs_refund_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode("redBookRefund");
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送小红书退款单数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                xhsRefundMapper.getPage(page, query);
            }
        }
        logger.info("小红书退款单修复完成");
    }

    private void fixXhsBill() {
        logger.info("开始修复小红书账单...");
        List<String> allTable = xhsBillMapper.getAllTable();
        Page<XhsBill> page;
        XhsBillQuery query;
        String dataId;
        MessageVo vo;
        for (String tableName : allTable) {
            logger.info("开始修复表：{}", tableName);
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new XhsBillQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            xhsBillMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (XhsBill xhsBill : page.getRecords()) {
                    dataId = xhsBill.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(xhsBill.getId());
                        vo.setStoreId(xhsBill.getStoreId());
                        vo.setBillMonth(xhsBill.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("xhs_bill_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode(PlatFormEnum.XHS_CONFIG.getValue());
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送小红书账单数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                xhsBillMapper.getPage(page, query);
            }
        }
        logger.info("小红书账单修复完成");
    }

    private void fixWmFundFlow() {
        logger.info("开始修复微盟资金流水...");
        List<String> allTable = wmFundFlowMapper.getAllTable();
        Page<WmFundFlow> page;
        WmFundFlowQuery query;
        String dataId;
        MessageVo vo;
        for (String tableName : allTable) {
            logger.info("开始修复表：{}", tableName);
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new WmFundFlowQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            wmFundFlowMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (WmFundFlow wmFundFlow : page.getRecords()) {
                    dataId = wmFundFlow.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(wmFundFlow.getId());
                        vo.setStoreId(wmFundFlow.getStoreId());
                        vo.setBillMonth(wmFundFlow.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("wm_fund_flow_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode("wmFundFlow");
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送微盟资金流水数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                wmFundFlowMapper.getPage(page, query);
            }
        }
        logger.info("修复微盟资金流水完成");
    }

    private void fixWdFundFlow() {
        logger.info("开始修复微店资金流水...");
        List<String> allTable = wdFundFlowMapper.getAllTable();
        Page<WdFundFlow> page;
        WdFundFlowQuery query;
        String dataId;
        MessageVo vo;
        for (String tableName : allTable) {
            logger.info("开始修复表：{}", tableName);
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new WdFundFlowQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            wdFundFlowMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (WdFundFlow wdFundFlow : page.getRecords()) {
                    dataId = wdFundFlow.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(wdFundFlow.getId());
                        vo.setStoreId(wdFundFlow.getStoreId());
                        vo.setBillMonth(wdFundFlow.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("wd_fund_flow_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode("wdFundFlow");
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送微店资金流水数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                wdFundFlowMapper.getPage(page, query);
            }
        }
        logger.info("修复微店资金流水完成");
    }

    private void fixTmFundFlow() {
        logger.info("开始修复天猫资金流水...");
        List<String> allTable = tmFundFlowMapper.getAllTable();
        Page<TmFundFlow> page;
        TmFundFlowQuery query;
        String dataId;
        MessageVo vo;
        for (String tableName : allTable) {
            logger.info("开始修复表：{}", tableName);
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new TmFundFlowQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            tmFundFlowMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (TmFundFlow tmFundFlow : page.getRecords()) {
                    dataId = tmFundFlow.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(tmFundFlow.getId());
                        vo.setStoreId(tmFundFlow.getStoreId());
                        vo.setBillMonth(tmFundFlow.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("tm_fund_flow_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode("tmFundFlow");
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送天猫资金流水数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                tmFundFlowMapper.getPage(page, query);
            }
        }
        logger.info("修复天猫资金流水完成");
    }

    private void fixTmBill() {
        logger.info("开始修复天猫账单明细...");
        List<String> allTable = tmBillMapper.getAllTable();
        Page<TmBill> page;
        TmBillQuery query;
        String dataId;
        MessageVo vo;
        for (String tableName : allTable) {
            logger.info("开始修复表：{}", tableName);
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new TmBillQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            tmBillMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (TmBill tmBill : page.getRecords()) {
                    dataId = tmBill.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(tmBill.getId());
                        vo.setStoreId(tmBill.getStoreId());
                        vo.setBillMonth(tmBill.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("tm_bill_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode(PlatFormEnum.TMALL_CONFIG.getValue());
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送天猫账单明细数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                tmBillMapper.getPage(page, query);
            }
        }
        logger.info("修复天猫账单明细完成");
    }

    private void fixPddFundFlow() {
        logger.info("开始修复拼多多资金流水明细...");
        List<String> allTable = pddFundFlowMapper.getAllTable();
        Page<PddFundFlow> page;
        PddFundFlowQuery query;
        String dataId;
        for (String tableName : allTable) {
            logger.info("开始修复表：{}", tableName);
            MessageVo vo;
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new PddFundFlowQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            pddFundFlowMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (PddFundFlow pddFundFlow : page.getRecords()) {
                    dataId = pddFundFlow.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(pddFundFlow.getId());
                        vo.setStoreId(pddFundFlow.getStoreId());
                        vo.setBillMonth(pddFundFlow.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("pdd_fund_flow_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode("pddFundFlow");
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送拼多多资金流水明细数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                pddFundFlowMapper.getPage(page, query);
            }
        }
        logger.info("修复拼多多资金流水明细完成");
    }

    private void fixPddDeposit() {
        logger.info("开始修复拼多多保证金明细...");
        List<String> allTable = pddDepositMapper.getAllTable();
        Page<PddDeposit> page;
        PddDepositQuery query;
        String dataId;
        for (String tableName : allTable) {
            logger.info("开始修复表：{}", tableName);
            MessageVo vo;
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new PddDepositQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            pddDepositMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (PddDeposit pddDeposit : page.getRecords()) {
                    dataId = pddDeposit.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(pddDeposit.getId());
                        vo.setStoreId(pddDeposit.getStoreId());
                        vo.setBillMonth(pddDeposit.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("pdd_deposit_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode("pddDeposit");
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送拼多多保证金明细数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                pddDepositMapper.getPage(page, query);
            }
        }
        logger.info("修复拼多多保证金明细完成...");
    }

    private void fixJdFundFlow() {
        logger.info("开始修复京东资金流水...");
        List<String> allTable = jdFundFlowMapper.getAllTable();
        Page<JdFundFlow> page;
        JdFundFlowQuery query;
        String dataId;
        for (String tableName : allTable) {
            logger.info("开始修复表：{}", tableName);
            MessageVo vo;
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new JdFundFlowQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            jdFundFlowMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (JdFundFlow jdFundFlow : page.getRecords()) {
                    dataId = jdFundFlow.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(jdFundFlow.getId());
                        vo.setStoreId(jdFundFlow.getStoreId());
                        vo.setBillMonth(jdFundFlow.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("jd_fund_flow_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode("jdFundFlow");
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送京东资金流水数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                jdFundFlowMapper.getPage(page, query);
            }
        }
        logger.info("修复京东资金流水完成...");
    }

    private void fixJdBill() {
        logger.info("开始修复京东账单...");
        List<String> allTable = jdBillMapper.getAllTable();
        Page<JdBill> page;
        JdBillQuery query;
        String dataId;
        for (String tableName : allTable) {
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new JdBillQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            logger.info("开始修复表：{}", tableName);
            MessageVo vo;
            jdBillMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (JdBill jdBill : page.getRecords()) {
                    dataId = jdBill.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(jdBill.getId());
                        vo.setStoreId(jdBill.getStoreId());
                        vo.setBillMonth(jdBill.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("jd_bill_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode(PlatFormEnum.JD_CONFIG.getValue());
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送京东账单数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                jdBillMapper.getPage(page, query);
            }
        }
        logger.info("京东账单修复完毕...");
    }

    private void fixDyFundFlow() {
        logger.info("开始修复抖音资金流水...");
        List<String> allTable = dyFundFlowMapper.getAllTable();
        Page<DyFundFlow> page;
        DyFundFlowQuery query;
        String dataId;
        for (String tableName : allTable) {
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new DyFundFlowQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            logger.info("开始修复表：{}", tableName);
            MessageVo vo;
            dyFundFlowMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (DyFundFlow dyFundFlow : page.getRecords()) {
                    dataId = dyFundFlow.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(dyFundFlow.getId());
                        vo.setStoreId(dyFundFlow.getStoreId());
                        vo.setBillMonth(dyFundFlow.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("dy_fund_flow_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode("dyFundFlow");
                        // 推送到转换标准账单的消息队列
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送抖音资金流水数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                dyFundFlowMapper.getPage(page, query);
            }
        }
        logger.info("抖音资金流水修复完成...");
    }

    private void fixDyBill() {
        logger.info("开始修复抖音账单...");
        List<String> allTable = dyBillMapper.getAllTable();
        Page<DyBill> page;
        DyBillQuery query;
        String dataId;
        for (String tableName : allTable) {
            page = new Page<>();
            page.setSize(100);
            page.setCurrent(1);
            query = new DyBillQuery();
            query.setTableNameList(Collections.singletonList(tableName));
            logger.info("开始修复表：{}", tableName);
            MessageVo vo;
            dyBillMapper.getPage(page, query);
            while (page.getRecords().size() > 0) {
                for (DyBill dyBill : page.getRecords()) {
                    dataId = dyBill.getId();
                    if (kafkaConsumeRecordService.checkCanConsume(dataId)) {
                        logger.info("可以消费：{}", dataId);
                        vo = new MessageVo();
                        vo.setDataId(dyBill.getId());
                        vo.setStoreId(dyBill.getStoreId());
                        vo.setBillMonth(dyBill.getId().substring(0, 6));
                        // 根据字典中的备注查找对应的数据库表字段名
                        vo.setTableNamePrefix("dy_bill_");
                        // 从字段管理里找到对应的字段类型编码，获取对应的billType
                        vo.setDictCode(PlatFormEnum.DY_CONFIG.getValue());
                        vo.setUuid(snowflakeIdWorker.nextId());
                        CompletableFuture<SendResult<String, String>> completableFuture = kafkaTemplate.send(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE, JSONObject.toJSONString(vo));
                        MessageVo finalVo = vo;
                        completableFuture.whenComplete((result, ex) -> {
                            if (Objects.nonNull(ex)) {
                                ex.printStackTrace();
                                logger.info("推送抖音账单数据到消息队列失败，失败原因：{}", ex.getMessage());
                                KafkaSendErrorRecord errorRecord = new KafkaSendErrorRecord();
                                errorRecord.setDataText(JSONObject.toJSONString(finalVo));
                                errorRecord.setQueueName(KafkaConstants.HWG_TOPIC_FIX_BILL_QUEUE);
                                errorRecord.setRemark(ex.getMessage());
                                errorRecord.setUuid(finalVo.getUuid());
                                kafkaSendErrorRecordService.checkAndSave(errorRecord);
                            }
                        });
                    }
                }
                try {
                    Thread.sleep(30000);
                    logger.info("休眠30秒");
                } catch (Exception e) {
                }
                page.setCurrent(page.getCurrent() + 1);
                dyBillMapper.getPage(page, query);
            }
        }
        logger.info("抖音账单修复完成...");
    }


}
